package com.kafka;

import java.lang.reflect.Type;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

import com.common.constant.CommonConstant;
import com.common.constant.DictConstant;
import com.common.constant.RedisKeys;
import com.common.dto.SiteTypeDetailsDto;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.util.DateUtil;
import com.util.JedisUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.util.CsvUtil;
import org.springframework.util.ObjectUtils;
import redis.clients.jedis.Jedis;
import scala.util.parsing.json.JSONObject;

/**
 * 接受kafka数据
 *
 * @author admin
 * 配置文件位置/usr/local/cp/cp_alarm_process/sys.properties
 * 本地F:/usr/local/cp/cp_alarm_process/sys.properties
 */
public class ReceiveKafkaData {
    private static KafkaConsumer<String, String> consumer;
    private final static String Kafka_Server_Ip = CsvUtil.getSysProperty("kafka.servers");
    //    private final static String []Device_No_List = new String[]{"video_alarm"};
    private final static String[] Device_No_List = new String[]{CsvUtil.getSysProperty("kafka.topic")};       //topic
    private final static String groupId = CsvUtil.formatDate(new Date(), CsvUtil.DATE_FORMAT7);// CsvUtil.getSysProperty("group.id.cp_alarm_process");
    final static Logger logger = LoggerFactory.getLogger(ReceiveKafkaData.class);
    private final static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
    private final static SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy/MM/dd/");

    //设备号与时间分别是0,1
    private final static Integer beginIndex = 2;

    //启动
    public static void main(String[] args) {//此方法需要的
        new ReceiveKafkaData().consume();
    }

    public ReceiveKafkaData() {
        logger.info("*********************开始初始化kafka消费者端*************************");
        Properties props = new Properties();
//        props.put("bootstrap.servers", "47.106.236.114:9092");
        props.put("bootstrap.servers", Kafka_Server_Ip);
        //每个消费者分配独立的组号
        props.put("group.id", groupId);//接受同一份消息group必须不一样

        //如果value合法，则自动提交偏移量
        props.put("enable.auto.commit", "true");
        //设置多久一次更新被消费消息的偏移量
        props.put("auto.commit.interval.ms", "1000");
        //设置会话响应的时间，超过这个时间kafka可以选择放弃消费或者消费下一条消息
        props.put("session.timeout.ms", "30000");
        //自动重置offset
        //props.put("auto.offset.reset","earliest");//当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
        props.put("auto.offset.reset", "latest");//当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据

        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<String, String>(props);
    }

    public void consume() {
        consumer.subscribe(Arrays.asList(Device_No_List));
        logger.info("开始消费topic: " + Device_No_List[0] + "");


        //消费原站点数据  保存到redis  保存到csv   处理后保存到warehouse_avg
        while (true) {
            //消费消息，将生产后的温度数据再放到kafka
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, String> record : records) {
                //String key = record.key();//消息的的key
                Jedis jedis = null;
                Gson gson = CommonConstant.gson;
                try {
                    long l = System.currentTimeMillis();
                    jedis = JedisUtil.getJedisPool().getResource(); //获取jedis资源

                    String gsonStr = record.value(); //在这里解析，不然后面每个地方都要解析
                    Type type = new TypeToken<List<List<String>>>() {}.getType(); //温度数据转list
                    List<List<String>> dataList = gson.fromJson(gsonStr, type); //15s内的温度数据集合(大概700毫秒一条)
                    //[000207, 1632254172782, 31.51, 31.87, 31.58, 31.97, 31.26, 31.43, 31.31, 31.11, 31.56, 31.46, 31.49, 31.17, 31.65, 31.66, 31.64, 31.23, null, null,等]
                    for (List<String> tempList : dataList){
                        /************ redis *******************************************************************************************************/
                        logger.info("将实时温度数据写入到redis");
                        System.out.println("解析list：" + tempList);
                        String deviceNo = tempList.get(0);

                        //Redis Lpush 命令将一个或多个值插入到列表头部。 如果 key 不存在，
                        // 一个空列表会被创建并执行 LPUSH 操作。 当 key 存在但不是列表类型时，返回一个错误
                        //这个用电池仓告警里面的升温告警那里 DEVICE_TEMP_DATA
                        //key="device:deviceNos_all" 保存实时温度数据, 数据格式：[{站点编号},{采集时间},{点1数据},{点2数据},.....{点n数据}]
                        jedis.lpush(String.format(RedisKeys.DEVICE_TEMP_DATA, deviceNo), gson.toJson(tempList).replace("\\\"", ""));

                        //这个用于breakdown程序的redis队列监听，发布主题breakdown
                        jedis.publish("breakdown", deviceNo);

                        //保存到redi，电池仓，充电柜，线槽，redis保存日报表一份，单独再保存一份，保存两份，key不同
                        //保存极值温度数据, 数据格式：[{站点编号},{采集时间},{max},{avg},{min}]###########################
                        saveExtremumData(tempList, deviceNo, jedis);//如果main项目没启动，缓存到redis中通道的信息，通道类型不存在，不保存极值数据，就不保存


                        /************* csv ********************************************************************************************************/
                        logger.info("将原始实时温度数据写入到csv");
                        //保存的文件路径 + 文件名 + yyyy-MM-dd
                        //保存的文件路径+ /年/月/日 + 站点编号
                        //mnt/site_data/2021/12/04/deviceNo_000207
                        //String fileName = CsvUtil.getSysProperty("cvs.path") + "device_" + deviceNo + "_" + DateUtil.formatOnlyDate(new Date(Long.parseLong(date)));
                        String fileName = CsvUtil.getSysProperty("cvs.path")+sdf2.format(new Date())+"deviceNo_"+deviceNo;
                        CsvUtil.writeFileForCsv(fileName, gson.toJson(tempList).split(","));


                        //处理电池仓层平均温度#################################################################################################################
                        //保存到redis   "warehouse_avg:%s，这个用于电池仓的告警程序"和csv  电池仓  4层
                        saveWarehouseAverageTemps(tempList,jedis);

                        logger.info("接受数据耗时：" + (System.currentTimeMillis() - l) + "ms");
                        logger.info("redis连接数：" + JedisUtil.getJedisPool().getNumActive());
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    if (jedis != null) {
                        //释放资源
                        JedisUtil.getJedisPool().returnResource(jedis);
                    }
                }


            }
        }
    }


    /**
     * 保存温度极值数据
     **/
    private void saveExtremumData(List<String> dataList, String deviceNo, Jedis jedis) {
        logger.info("开始处理极值数据--->");
        Gson gson = CommonConstant.gson;
        //通道总数(每个通道固定20个测温点)
        Integer len = (dataList.size() - 2) / 20;//56
        //"cache:site-type-id:%s" 从零开始截取56，SiteTypeDetailsDto每个通道的信息
        List<SiteTypeDetailsDto> aisleList = jedis.lrange(
                String.format(RedisKeys.SITE_TYPE_CACHE, len),
                0, len).stream().map(SiteTypeDetailsDto::convert).collect(Collectors.toList());
        //如果不存在该通道的缓存就直接跳过
        if (!jedis.exists(String.format(RedisKeys.SITE_TYPE_CACHE, len))) {
            logger.info(len + "通道类型不存在，不保存极值数据");
            return;
        }
        //获取电池仓的测温点总数  warehouse
        Integer warehouseNum = aisleList.stream().filter(d -> DictConstant.AisleType.warehouse.equals(d.getAisleType())).mapToInt(SiteTypeDetailsDto::getPointCount).sum();
        //获取充电柜的测温点总数
        Integer cabinetNum = aisleList.stream().filter(d -> DictConstant.AisleType.cabinet.equals(d.getAisleType())).mapToInt(SiteTypeDetailsDto::getPointCount).sum();
        //获取线槽的测温点总数
        Integer lineNum = aisleList.stream().filter(d -> DictConstant.AisleType.line.equals(d.getAisleType())).mapToInt(SiteTypeDetailsDto::getPointCount).sum();

        //获取电池仓的温度区间下标
        Integer warehouseIndex = warehouseNum + beginIndex;
        //获取充电柜的温度区间下标
        Integer cabinetIndex = warehouseIndex + cabinetNum;
        //获取线槽的测温度区间下标
        Integer lineIndex = cabinetIndex + lineNum;

        //截取电池仓温度数据
        List<BigDecimal> warehouseTempData = dataList.subList(beginIndex, warehouseIndex).stream().filter(StringUtils::isNotBlank).map(BigDecimal::new).collect(Collectors.toList());
        //截取充电柜温度数据
        List<BigDecimal> cabinetTempData = dataList.subList(warehouseIndex, cabinetIndex).stream().filter(StringUtils::isNotBlank).map(BigDecimal::new).collect(Collectors.toList());
        //截取线槽温度数据
        List<BigDecimal> lineTempData = dataList.subList(cabinetIndex, lineIndex).stream().filter(StringUtils::isNotBlank).map(BigDecimal::new).collect(Collectors.toList());

        List<String> extremumList = new ArrayList<>();
        BigDecimal max;
        BigDecimal avg;
        BigDecimal min;
        String timeStr = sdf.format(new Date());
        //保存电池仓极值数据
        if (warehouseTempData != null && warehouseTempData.size() != 0) {
            max = warehouseTempData.stream().distinct().max(BigDecimal::compareTo).get();
            avg = warehouseTempData.stream().reduce(BigDecimal.ZERO, BigDecimal::add).divide(BigDecimal.valueOf(warehouseTempData.size()), 2, BigDecimal.ROUND_DOWN);
            min = warehouseTempData.stream().distinct().min(BigDecimal::compareTo).get();
            ;
            extremumList.add(dataList.get(0));//站点
            extremumList.add(dataList.get(1));//时间
            extremumList.add(String.valueOf(max));
            extremumList.add(String.valueOf(avg));
            extremumList.add(String.valueOf(min));
            String json = gson.toJson(extremumList);
            jedis.lpush(String.format(RedisKeys.WAREHOUSE_TEMP_DATA, deviceNo), json);//"device:%s_warehouse"
            //保存到日报表缓存中
            jedis.lpush(String.format(RedisKeys.REPORT_WAREHOUSE, deviceNo,timeStr), json);//report:%s_%s_warehouse
        } else {
            logger.info("电池仓不存在温度数据，不保存电池仓极值");
        }

        if (cabinetTempData != null && cabinetTempData.size() != 0) {
            //保存充电柜极值数据
            extremumList = new ArrayList<>();
            max = cabinetTempData.stream().distinct().max(BigDecimal::compareTo).get();
            avg = cabinetTempData.stream().reduce(BigDecimal.ZERO, BigDecimal::add).divide(BigDecimal.valueOf(cabinetTempData.size()), 2, BigDecimal.ROUND_DOWN);
            min = cabinetTempData.stream().distinct().min(BigDecimal::compareTo).get();
            ;
            extremumList.add(dataList.get(0));
            extremumList.add(dataList.get(1));
            extremumList.add(String.valueOf(max));
            extremumList.add(String.valueOf(avg));
            extremumList.add(String.valueOf(min));
            String json = gson.toJson(extremumList);
            jedis.lpush(String.format(RedisKeys.CABINET_TEMP_DATA, deviceNo), json);
            //保存到日报表缓存中
            jedis.lpush(String.format(RedisKeys.REPORT_CABINET, deviceNo,timeStr), json);
        } else {
            logger.info("充电柜不存在温度数据，不保存充电柜极值");
        }

        if (lineTempData != null && lineTempData.size() != 0) {
            //保存线槽极值数据
            extremumList = new ArrayList<>();
            max = lineTempData.stream().distinct().max(BigDecimal::compareTo).get();
            avg = lineTempData.stream().reduce(BigDecimal.ZERO, BigDecimal::add).divide(BigDecimal.valueOf(lineTempData.size()), 2, BigDecimal.ROUND_DOWN);
            min = lineTempData.stream().distinct().min(BigDecimal::compareTo).get();

            extremumList.add(dataList.get(0));
            extremumList.add(dataList.get(1));
            extremumList.add(String.valueOf(max));
            extremumList.add(String.valueOf(avg));
            extremumList.add(String.valueOf(min));
            String json = gson.toJson(extremumList);
            jedis.lpush(String.format(RedisKeys.LINE_TEMP_DATA, deviceNo), json);
            //保存到日报表缓存中
            jedis.lpush(String.format(RedisKeys.REPORT_LINE, deviceNo,timeStr), json);
        } else {
            logger.info("线槽不存在温度数据，不保存线槽极值");
        }
    }


    /**
     * @Desc: 电池仓层平均温处理
     **/
    private void saveWarehouseAverageTemps(List<String> dataList, Jedis jedis){

        String siteNo = dataList.get(0);
        String date = dataList.get(1);

        List<String> temps = dataList.subList(2, dataList.size());
        /****
         * 第一层 5/9/13/17/21/25
         * 第二次 6/10/14/18/22/26
         * 第三层 7/11/15/19/23/27
         * 第四层 8/12/16/20/24/28
         */
        List<String> result = new ArrayList<>();
        result.add(dataList.get(1));

        for (int level = 0; level < CommonConstant.C_.length; level++) {
            //搜集当前层所有通道的平均值
            List<Double> aisleAvgs = new ArrayList<>();
            for (int aisleIndex = 0; aisleIndex < CommonConstant.C_[level].length; aisleIndex++) {
                int aisleNo = CommonConstant.C_[level][aisleIndex];//通道号码
                //每个通道有20个测温点
                int startIndex = 20*(aisleNo-1);
                int endIndex = startIndex+19;
                //求得单个通道平均值
                List<String> currentAisleTemps = temps.subList(startIndex, endIndex + 1)
                        .stream()
                        .filter(v->!ObjectUtils.isEmpty(v)).collect(Collectors.toList());

                if(currentAisleTemps.size()>0){
                    double average = currentAisleTemps
                            .stream()
                            .mapToDouble(Double::parseDouble)
                            .average().getAsDouble();
                    aisleAvgs.add(average);//添加每个通道的平均值进去
                }else{
                    aisleAvgs.add(0.0);
                }

            }
            //求出当前层的平均值  例如第一层6个通道的平均值
            List<Double> levelAvgs = aisleAvgs.stream().filter(v->!v.equals(0.0)).collect(Collectors.toList());
            if(levelAvgs.size()>0){
                double levelAvg = levelAvgs.stream().mapToDouble(value -> value)
                        .average().getAsDouble();
                result.add(String.format("%.2f", levelAvg));////添加每一层的平均值进去
            }else{
                result.add("0.0");
            }

        }

        String reJson = CommonConstant.gson.toJson(result);

        // //这个用电池仓告警里面的升温告警哪里 DEVICE_TEMP_DATA
        //存储到缓存   站点电池仓各层平均温    "warehouse_avg:%s"    从头开始放
        jedis.lpush(String.format(RedisKeys.WAREHOUSE_AVG_LEVEL,siteNo),reJson);

        //存储到csv  D:\mnt\site_data\warehouse_avg\device_000207_2021-09-22
        String fileName = CsvUtil.getSysProperty("cvs.path")+ "warehouse_avg/" + "device_" + siteNo + "_" + DateUtil.formatOnlyDate(new Date(Long.parseLong(date)));
        try {
            CsvUtil.writeFileForCsv(fileName, reJson.split(","));
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * @Desc: 获取站点文件全路径
     **/
    private static String getFileName(String siteNo){
        String ymr = new SimpleDateFormat("yyyy/MM/dd/").format(new Date());
        return CsvUtil.getSysProperty("cvs.path") + ymr + siteNo;
    }

}
